/* This class implements the optimizer in query processing. The algorithm implemented here is mostly similar to the System R*'s algorithm.
 * It first generates all operation trees by permutation of the join ordering.
 * For each join operation within a tree, it selects a join algorithm based on database statistics.
 * Among these trees, the one with the lowest cost is chosen for further optimization.
 * Then in this optimal tree, for each join operation, it decides the site that does the actual join
 * and the method of sending data between sites. As described in the text book, these are total four strategies to consider.
 * It evaluates these four and picks the lowest-cost strategy.
 * 
 * (c) Hongbin Kuang
 * 04/23/2009
 * V1.0
 */

package costEstimation;

import java.util.*;

import queryTree.*;
import optimization.*;

public class Optimizer2{
	private StatList allStat = new StatList(); // List of database statistics
	private ArrayList<Field> topProjection; // The final projection of a query
	private ArrayList<Condition> joinConditions = new ArrayList<Condition>(); // All join conditions in a query
	private ArrayList<TableNode> tables = new ArrayList<TableNode>(); // All tables involved
	private ArrayList<CandidateTree> trees = new ArrayList<CandidateTree>(); // Trees generated by permutation
	private final int MERGE_JOIN = 0;
	private final int NESTED_LOOP = 1;
	private int perferredAlgorithm = MERGE_JOIN; // Use this algorithm when can't choose clearly
	
	// To reduce interaction between modules, supply initial statistics in construction function.
	public Optimizer2()
	{
		allStat.addStat(Stat.LT, 1);
		// CT same as LT because LT implicitly includes CT in our experiment -- 
		// results are transmitted from Oracle server to test machines.
		allStat.addStat(Stat.CT, 1);
		allStat.addStat(Stat.LENGTH+"supplies-item", 10);
		allStat.addStat(Stat.LENGTH+"supplies-unitquantity", 8);
		allStat.addStat(Stat.LENGTH+"supplies-price", 8);
		allStat.addStat(Stat.LENGTH+"recipe-menu", 10);
		allStat.addStat(Stat.LENGTH+"recipe-ingredient", 10);
		allStat.addStat(Stat.LENGTH+"recipe-servingquantity", 8);
		allStat.addStat(Stat.LENGTH+"budget-mealtime", 10);
		allStat.addStat(Stat.LENGTH+"budget-day", 8);
		allStat.addStat(Stat.LENGTH+"budget-budgetleft", 8);
		allStat.addStat(Stat.CARD_DOM+"supplies-item", 10000);
		allStat.addStat(Stat.CARD_DOM+"supplies-unitquantity", 1000);
		allStat.addStat(Stat.CARD_DOM+"supplies-price", 5000);
		allStat.addStat(Stat.CARD_DOM+"recipe-menu", 1000);
		allStat.addStat(Stat.CARD_DOM+"recipe-ingredient", 10000);
		allStat.addStat(Stat.CARD_DOM+"recipe-servingquantity", 1000);
		allStat.addStat(Stat.CARD_DOM+"budget-mealtime", 100);
		allStat.addStat(Stat.CARD_DOM+"budget-day", 3000);
		allStat.addStat(Stat.CARD_DOM+"budget-budgetleft", 10000);
		allStat.addStat(Stat.CARD_TABLE+"supplies", 3000);
		allStat.addStat(Stat.CARD_TABLE+"recipe", 3000);
		allStat.addStat(Stat.CARD_TABLE+"budget", 3000);
		allStat.addStat(Stat.CARD_ATTR+"supplies-item", 300);
		allStat.addStat(Stat.CARD_ATTR+"supplies-unitquantity", 300);
		allStat.addStat(Stat.CARD_ATTR+"supplies-price", 300);
		allStat.addStat(Stat.CARD_ATTR+"recipe-menu", 300);
		allStat.addStat(Stat.CARD_ATTR+"recipe-ingredient", 300);
		allStat.addStat(Stat.CARD_ATTR+"recipe-servingquantity", 300);
		allStat.addStat(Stat.CARD_ATTR+"budget-mealtime", 300);
		allStat.addStat(Stat.CARD_ATTR+"budget-day", 300);
		allStat.addStat(Stat.CARD_ATTR+"budget-budgetleft", 300);
		// Table names in the join is ordered lexicographically
		allStat.addStat(Stat.JOIN_SELECT+"recipe-supplies", 0.01);
		allStat.addStat(Stat.JOIN_SELECT+"budget-supplies", 0.01);
		allStat.addStat(Stat.JOIN_SELECT+"budget-recipe", 0.01);
		// Only have MAX and MIN for numerical attributes,
		// assuming selects on non-numerical attributes are all "equi-select"
		// thus don't need MAX and MIN for estimation.
		allStat.addStat(Stat.MAX+"supplies-unitquantity", 1000);
		allStat.addStat(Stat.MAX+"supplies-price", 1000);
		allStat.addStat(Stat.MAX+"recipe-servingquantity", 1000);
		allStat.addStat(Stat.MAX+"budget-day", 21000000);
		allStat.addStat(Stat.MAX+"budget-budgetleft", 10000);
		allStat.addStat(Stat.MIN+"supplies-unitquantity", 0);
		allStat.addStat(Stat.MIN+"supplies-price", 0);
		allStat.addStat(Stat.MIN+"recipe-servingquantity", 0);
		allStat.addStat(Stat.MIN+"budget-day", 19000000);
		allStat.addStat(Stat.MIN+"budget-budgetleft", -100);		
	}
	
	// The core function for optimization
	public AbstractNode optimization(AbstractNode root)
	{
		if (root instanceof JoinNode)
		{
			topProjection = root.getAttrsProjection();
			readTablesAndConditions(root); // Record all tables and join conditions in the query for other functions
			AbstractNode optimalTree = permutation(); // Generate all trees and select the optimal one
			joinStrategy(optimalTree); // Assign a join strategy for each join operation
			return optimalTree;
		}
		else
			return root; // No join operations
	}

	// After each join operation, the system calls this function to update join selectivity factors
	public boolean updateJoinStat(String tableA, String tableB, int tuples)
	{
		// Order join table names lexicographically
		String name;
		if (tableA.compareToIgnoreCase(tableB) <= 0)
		{
			name = Stat.JOIN_SELECT+tableA+"-"+tableB;
		}
		else
		{
			name = Stat.JOIN_SELECT+tableB+"-"+tableA;
		}
		
		String nameA = Stat.CARD_TABLE+tableA;
		String nameB = Stat.CARD_TABLE+tableB;
		double value = tuples/((allStat.getStat(nameA))*(allStat.getStat(nameB)));
		allStat.updateStat(name, value);
		
		return true;
	}
	
	// Record all tables and join conditions in the query
	private boolean readTablesAndConditions(AbstractNode node)
	{
		if (node instanceof JoinNode)
		{
			joinConditions.addAll(((JoinNode)node).getJoinConditionList());
			boolean a = readTablesAndConditions((AbstractNode)((JoinNode)node).getLeftJoinTree());
			boolean b = readTablesAndConditions((AbstractNode)((JoinNode)node).getRightJoinTree());
			return a && b;
		}
		else if (node instanceof TableNode)
		{
			tables.add((TableNode)node);
			return true;
		}
		
		return false;
	}
	
	// Generate all trees and select the optimal
	private AbstractNode permutation()
	{
		// First make each table as a tree
		for (int i=0; i<tables.size(); i++)
		{
			CandidateTree seed = new CandidateTree();
			seed.root = (AbstractNode)tables.get(i);
			seed.tablesToJoin.addAll(tables);
			seed.tablesToJoin.remove(i);
			
			trees.add(seed);
		}
		
		// Join one more table in each loop
		do
		{
			ArrayList<CandidateTree> nextLevelTrees = new ArrayList<CandidateTree>();
			// Scan each tree in the lower level and its remaining tables for join
			for (int i=0; i<trees.size(); i++)
			{
				for (int j=0; j<trees.get(0).tablesToJoin.size(); j++)
				{
					// For each pair of tree and table, there are two orderings by flipping
					JoinNode node1 = new JoinNode();
					JoinNode node2 = new JoinNode();
					CandidateTree tree1 = new CandidateTree();
					CandidateTree tree2 = new CandidateTree();
					tree1.root = (AbstractNode)node1;
					tree1.tablesToJoin.addAll(trees.get(i).tablesToJoin);
					tree2.root = (AbstractNode)node2;
					tree2.tablesToJoin.addAll(trees.get(i).tablesToJoin);

					node1.setLeftJoinTree((TreeNode)trees.get(i).tablesToJoin.get(j));
					node1.setRightJoinTree((TreeNode)trees.get(i).root);
					node2.setLeftJoinTree((TreeNode)trees.get(i).root);
					node2.setRightJoinTree((TreeNode)trees.get(i).tablesToJoin.get(j));
					tree1.tablesToJoin.remove(j);
					tree2.tablesToJoin.remove(j);
					
					// Record all tables involved in a join for other functions
					joinTables(node1);
					joinTables(node2);
					
					// Determine the join conditions
					joinConditions(node1);
					joinConditions(node2);
					
					// Choose a join algorithm
					joinAlgorithm(node1);
					joinAlgorithm(node2);
					
					nextLevelTrees.add(tree1);
					nextLevelTrees.add(tree2);
				}
			}
						
			trees = nextLevelTrees;
			pruneCandidate(); // Prune some candidate using heuristics
		}
		while(trees.get(0).tablesToJoin.size() > 0);
		
		trees.get(0).root.setAttrsProjection(topProjection);
		
		// All trees have same cost at this level. Only try the first tree for further estimation.
		return trees.get(0).root;
	}
	
	// Record all tables involved in a join
	private boolean joinTables(JoinNode node)
	{
		TreeNode left = node.getLeftJoinTree();
		TreeNode right = node.getRightJoinTree();
		
		if (left instanceof JoinNode)
		{
			node.tables.addAll(((JoinNode)left).tables);
		}
		else if (left instanceof TableNode)
		{
			node.tables.add(((TableNode)left).getTableName());
		}
		else
			return false;
		
		if (right instanceof JoinNode)
		{
			node.tables.addAll(((JoinNode)right).tables);
		}
		else if (right instanceof TableNode)
		{
			node.tables.add(((TableNode)right).getTableName());
		}
		else
			return false;
		
		return true;
	}

	// Determine the join conditions
	private boolean joinConditions(JoinNode node)
	{
		int k = joinConditions.size();
		for (int i=0; i<k; i++)
		{
			String left = mappingNames(joinConditions.get(i).lhs.table_name);
			String right = mappingNames(joinConditions.get(i).rhs.table_name);
			// If tables involved in a join include a condition's both left and right side, the join needs this condition
			if (node.tables.indexOf(left) != -1 && node.tables.indexOf(right) != -1) 
			{
				node.addJoinCondition(joinConditions.get(i));
				joinConditions.remove(i);
				i--;
				k--;
			}
		}
		
		return true;
	}
	
	// Mapping other team member's definition to Strings
	private String mappingNames(tab_name_att_name t)
	{
		if (t == tab_name_att_name.BUDGET)
			return new String("budget");
		else if (t == tab_name_att_name.RECIPE)
			return new String("recipe");
		else if (t == tab_name_att_name.SUPPLIES)
			return new String("supplies");
		else if (t == tab_name_att_name.ITEM)
			return new String("item");
		else if (t == tab_name_att_name.UNITQUANTITY)
			return new String("unitquantity");
		else if (t == tab_name_att_name.PRICE)
			return new String("price");
		else if (t == tab_name_att_name.MENU)
			return new String("menu");
		else if (t == tab_name_att_name.INGREDIENT)
			return new String("ingredient");
		else if (t == tab_name_att_name.SERVINGQUANTITY)
			return new String("servingquantity");
		else if (t == tab_name_att_name.MEALTIME)
			return new String("mealtime");
		else if (t == tab_name_att_name.DAY)
			return new String("day");
		else if (t == tab_name_att_name.BUDGETLEFT)
			return new String("budgetleft");
		
		return null;
	}
	
	// Choose a join algorithm
	private boolean joinAlgorithm(JoinNode node)
	{
		if (node.getJoinConditionList().size() == 0) // Cartesian product
		{
			node.joinAlgorithm = perferredAlgorithm;
			return true;
		}
		
		boolean equiJoin = true;
		
		for (int i=0; i<node.getJoinConditionList().size(); i++)
		{
			if (node.getJoinConditionList().get(i).operator != operator_type.EQUAL)
			{
				equiJoin = false;
				break;
			}
		}
		
		if (equiJoin)
			node.joinAlgorithm = MERGE_JOIN; // Equi-join prefers merge join algorithm (we have sorted tables)
		else
			node.joinAlgorithm = perferredAlgorithm;
		
		return true;
	}
	
	// Prune some candidate using heuristics
	private boolean pruneCandidate()
	{
		// Remove trees with Cartesian product -- no join condition
		int k = trees.size();
		for(int i=0; i<k; i++)
		{
			if (((JoinNode)trees.get(i).root).getJoinConditionList().size() == 0)
			{
				if (k>1) // In case Cartesian is needed
				{
					trees.remove(i);
					i--;
					k--;
				}
				else
					return true;
			}
		}
		
		if (k == 1) // Prune not necessary
			return true;
		
		// If two trees are equivalent, remove one
		// If commutatively equivalent, remove the one with higher cost
		for (int i=0; i<k-1; i++)
			for (int j=i+1; j<k; j++)
			{
				if (sameTrees(trees.get(i).root, trees.get(j).root))
				{
					trees.remove(j);
					k--;
					j--;
				}
				else if (commutativeTrees(trees.get(i).root, trees.get(j).root))
				{
					int r = compareCosts((JoinNode)trees.get(i).root, (JoinNode)trees.get(j).root);
					// Leave lowest. Prune all others.
					if (r > 0)
					{
						trees.remove(i);
						k--;
						i--;
						break; // Skip rest of j
					}
					else if (r < 0)
					{
						trees.remove(j);
						k--;
						j--;
					}
				}
			}
		
		return true;
	}
	
	// Test whether two trees are the same
	private boolean sameTrees(AbstractNode root1, AbstractNode root2)
	{
		if ((root1 instanceof TableNode) && (root2 instanceof TableNode))
			if (((TableNode)root1).getTableName().equals(((TableNode)root2).getTableName()))
					return true;
			
		if ((root1 instanceof JoinNode) && (root2 instanceof JoinNode))
		{
			boolean a = sameTrees((AbstractNode)((JoinNode)root1).getLeftJoinTree(), (AbstractNode)((JoinNode)root2).getLeftJoinTree());
			boolean b = sameTrees((AbstractNode)((JoinNode)root1).getRightJoinTree(), (AbstractNode)((JoinNode)root2).getRightJoinTree());
			return a && b;
		}
				
		return false;
	}
	
	// Test whether two trees are commutatively equivalent
	private boolean commutativeTrees(AbstractNode root1, AbstractNode root2)
	{
		boolean a = sameTrees((AbstractNode)((JoinNode)root1).getLeftJoinTree(), (AbstractNode)((JoinNode)root2).getRightJoinTree());
		boolean b = sameTrees((AbstractNode)((JoinNode)root1).getRightJoinTree(), (AbstractNode)((JoinNode)root2).getLeftJoinTree());
		return a && b;
	}
	
	// Compare the cost of two ordering
	private int compareCosts(JoinNode node1, JoinNode node2)
	{
		if ((node1.joinAlgorithm == MERGE_JOIN) && (node2.joinAlgorithm == MERGE_JOIN)) // Same if merge join
			return 0;
		
		if ((node1.joinAlgorithm == NESTED_LOOP) && (node2.joinAlgorithm == NESTED_LOOP)) // More tricky if nested-loop
		{
			if ((node1.getLeftJoinTree() instanceof TableNode) && (node1.getRightJoinTree() instanceof TableNode)) // Joins at lowest level
			{
				int a = selectedTuples((TableNode)node1.getLeftJoinTree());
				int b = selectedTuples((TableNode)node1.getRightJoinTree());
				if (a>b)
					return 1;
				else if (a<b)
					return -1;
			}
			// Joins at upper levels
			else if (node1.getRightJoinTree() instanceof TableNode)
			{
				return -1;
			}
			else if (node2.getRightJoinTree() instanceof TableNode)
			{
				return 1;
			}
		}
		
		return 0;
	}
	
	// Number of tuples after selection
	private int selectedTuples(TableNode node)
	{
		String name1 = Stat.CARD_TABLE+node.getTableName();
		int tableCardinality = (int)allStat.getStat(name1);
		
		if (node.getSelectPredicate() instanceof SimpleBoolExpr) // Simple predicates without AND, OR ...
		{
			SimpleBoolExpr predicate = (SimpleBoolExpr)node.getSelectPredicate();
			if (predicate.getOperator() == SimpleBoolExpr.EQ) // Equal select
			{
				String name2 = Stat.CARD_ATTR+node.getTableName()+"-"+mappingNames(predicate.getAttr().item_name);
				return (int)(tableCardinality/allStat.getStat(name2));
			}
			else
			{
				boolean a = (predicate.getOperator() == SimpleBoolExpr.GE);
				boolean b = (predicate.getOperator() == SimpleBoolExpr.GT);
				boolean c = (predicate.getOperator() == SimpleBoolExpr.LE);
				boolean d = (predicate.getOperator() == SimpleBoolExpr.LT);
				String nameMAX = Stat.MAX+node.getTableName()+"-"+mappingNames(predicate.getAttr().item_name);
				String nameMIN = Stat.MIN+node.getTableName()+"-"+mappingNames(predicate.getAttr().item_name);
				double max = allStat.getStat(nameMAX);
				double min = allStat.getStat(nameMIN);
				if (a || b) // > select
				{
					return (int)(tableCardinality*(max-(new Double(predicate.getValue())))/(max-min));					
				}
				else if (c || d) // < select
				{
					return (int)(tableCardinality*((new Double(predicate.getValue()))-min)/(max-min));
				}
			}
		}
		
		// All other kinds of select predicates
		return tableCardinality;
	}
	
	// Assign a join strategy for each join operation
	private boolean joinStrategy(AbstractNode root)
	{		
		// We traverse the tree to reach each join
		if (root instanceof JoinNode)
		{
			// Strategies 1-3, 4 is ignored since it's always the worst in the experiments
			int k = ((JoinNode)root).getJoinConditionList().size();
			if (k>1 || k==0) // Cartesian product or join on multiple attributes, simply decide
				((JoinNode)root).joinStrategy = 1;
			else
			{
				double score1, score2;
				tab_name_att_name tableA = ((JoinNode)root).getJoinConditionList().get(0).lhs.table_name;
				tab_name_att_name tableB = ((JoinNode)root).getJoinConditionList().get(0).rhs.table_name;
				tab_name_att_name itemA = ((JoinNode)root).getJoinConditionList().get(0).lhs.item_name;
				tab_name_att_name itemB = ((JoinNode)root).getJoinConditionList().get(0).rhs.item_name;
				String ATTR = Stat.CARD_ATTR+mappingNames(tableA)+"-"+mappingNames(itemA);
				String DOM = Stat.CARD_DOM+mappingNames(tableA)+"-"+mappingNames(itemA);
				String TABLE1 = Stat.CARD_TABLE+mappingNames(tableA);
				String TABLE2 = Stat.CARD_TABLE+mappingNames(tableB);
				String nameLENGTH1 = Stat.LENGTH+mappingNames(tableA)+"-"+mappingNames(itemA);
				String nameLENGTH2 = Stat.LENGTH+mappingNames(tableB)+"-"+mappingNames(itemB);
				String nameLT = Stat.LT;
				String nameCT = Stat.CT;
				double lt = allStat.getStat(nameLT);
				double ct = allStat.getStat(nameCT);
				double tableCardinality1 = allStat.getStat(TABLE1);
				double tableCardinality2 = allStat.getStat(TABLE2);
				double length1 = allStat.getStat(nameLENGTH1);
				double length2 = allStat.getStat(nameLENGTH2);
				double s = allStat.getStat(ATTR)/allStat.getStat(DOM)*tableCardinality2/tableCardinality1;
				
				// Cost of strategy 1 and 2
				score1 = lt*tableCardinality1+lt*s*tableCardinality1+ct*tableCardinality1*length1*3;
				score2 = lt*(tableCardinality2*2+tableCardinality1)+lt*s*tableCardinality1+ct*tableCardinality2*length2*3;
				
				if (score1 > score2)
				{
					((JoinNode)root).joinStrategy = 2;
				}
				else
				{
					((JoinNode)root).joinStrategy = 1;
				}
				
				// Strategy only suitable for nested-loop
				if (((JoinNode)root).joinAlgorithm == NESTED_LOOP)
				{
					double score3 = lt*tableCardinality1+lt*s*tableCardinality1+ct*length1*tableCardinality1+ct*s*length2*3*tableCardinality1;
					
					if (((JoinNode)root).joinStrategy == 2)
						if (score3 < score2)
							((JoinNode)root).joinStrategy = 3;
					else if (((JoinNode)root).joinStrategy == 1)
						if (score3 < score1)
							((JoinNode)root).joinStrategy = 3;
				}
			}
			
			boolean a = joinStrategy((AbstractNode)((JoinNode)root).getLeftJoinTree());
			boolean b = joinStrategy((AbstractNode)((JoinNode)root).getRightJoinTree());
			return a && b;				
		}
		else if (root instanceof TableNode)
			return true;
		
		return false;
	}		
}